package com.dps.recommend.third;

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.model.*;
import com.dps.metrics.processor.CustomizedMetricsProcessor;
import com.dps.recommend.beans.DpsRequest;
import com.dps.recommend.utils.LogUtils;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import lombok.SneakyThrows;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * AliTableUtils 工具
 */
public class DpsAliTableUtils {

    private SyncClient client;

    private String env;

    public DpsAliTableUtils(SyncClient syncClient, String env) {
        this.client = syncClient;
        this.env = getTableStoreNameSuffix(env);
    }


    private String getTableStoreNameSuffix(String env) {
        if (StringUtils.isEmpty(env)) {
            env = "_test";
        } else if (env.equals("prod")) {
            env = "";
        } else {
            env = "_" + env;
        }
        return env;
    }


    /**
     * 批量获取
     *
     * @param dpsRequest
     * @param tableName
     * @param primaryKeyName
     * @param itemIds
     * @param client
     * @return
     */
    @SneakyThrows
    public Map<String, Map<String, String>> batchGetRow(DpsRequest dpsRequest, String tableName, String primaryKeyName, List<String> itemIds) {
        CustomizedMetricsProcessor processor = CustomizedMetricsProcessor.builder("dps_sence_layer_ts").tagsMap(getTsLayerTagMap(dpsRequest, tableName)).build();
        return processor.process(() -> {
            Stopwatch timer = Stopwatch.createStarted();
            Map<String, Map<String, String>> resMap = new HashMap<>();
            try {
                String tableNameStr = tableName + env;
                List<String> pkValueList = itemIds;
                if (CollectionUtils.isEmpty(pkValueList)) {
                    pkValueList = Lists.newArrayList();
                    return resMap;
                }
                MultiRowQueryCriteria multiRowQueryCriteria = new MultiRowQueryCriteria(tableNameStr);
                for (String itemId : pkValueList) {
                    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
                    primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName, PrimaryKeyValue.fromString(itemId));
                    PrimaryKey primaryKey = primaryKeyBuilder.build();
                    multiRowQueryCriteria.addRow(primaryKey);
                }
                multiRowQueryCriteria.setMaxVersions(1);
                BatchGetRowRequest batchGetRowRequest = new BatchGetRowRequest();
                batchGetRowRequest.addMultiRowQueryCriteria(multiRowQueryCriteria);
                BatchGetRowResponse batchGetRowResponse = client.batchGetRow(batchGetRowRequest);
                // LogUtils.debug(dpsRequest, String.format("is all success:%s", batchGetRowResponse.isAllSucceed()));
                parseBatchRowRes(dpsRequest, tableNameStr, primaryKeyName, resMap, batchGetRowResponse);
                if (!batchGetRowResponse.isAllSucceed()) {
                    for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getFailedRows()) {
                        LogUtils.debug(dpsRequest, String.format("failure line：%s", batchGetRowRequest.getPrimaryKey(rowResult.getTableName(), rowResult.getIndex())));
                        LogUtils.debug(dpsRequest, String.format("failure reason：%s", rowResult.getError()));

                    }
                    //重试
                    BatchGetRowRequest retryRequest = batchGetRowRequest.createRequestForRetry(batchGetRowResponse.getFailedRows());
                    BatchGetRowResponse batchGetRowResponseAgain = client.batchGetRow(retryRequest);
                    parseBatchRowRes(dpsRequest, tableNameStr, primaryKeyName, resMap, batchGetRowResponseAgain);
                }
            } catch (Exception e) {
                LogUtils.error(dpsRequest, String.format("batchGetRow,tableName:%s, primaryKeyName:%s occurs error", tableName, primaryKeyName, e));
            } finally {
                timer.stop();
                LogUtils.info(dpsRequest, String.format("tableName=%s,primaryKeyName=%s,pkValueSize=%s,time=%s", tableName, primaryKeyName, itemIds.size(), timer.elapsed(TimeUnit.MILLISECONDS)));
            }
            return resMap;
        });

    }


    /**
     * 批量获取指定列名
     *
     * @param dpsRequest
     * @param tableName
     * @param primaryKeyName
     * @param pkValues
     * @param columns
     * @param client
     * @return
     */
    public Map<String, Map<String, String>> batchGetRowColumns(DpsRequest dpsRequest, String tableName, String primaryKeyName,
                                                               List<String> pkValues, List<String> columns) {
        tableName = tableName + env;
        Map<String, Map<String, String>> resMap = new HashMap<>();
        if (CollectionUtils.isEmpty(pkValues)) return resMap;
        try {
            MultiRowQueryCriteria multiRowQueryCriteria = new MultiRowQueryCriteria(tableName);
            for (String itemId : pkValues) {
                PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
                primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName, PrimaryKeyValue.fromString(itemId));
                PrimaryKey primaryKey = primaryKeyBuilder.build();
                multiRowQueryCriteria.addRow(primaryKey);
            }
            multiRowQueryCriteria.setMaxVersions(1);
            for (String column : columns) {
                multiRowQueryCriteria.addColumnsToGet(column);
            }
            BatchGetRowRequest batchGetRowRequest = new BatchGetRowRequest();
            batchGetRowRequest.addMultiRowQueryCriteria(multiRowQueryCriteria);
            BatchGetRowResponse batchGetRowResponse = client.batchGetRow(batchGetRowRequest);
            //LogUtils.debug(dpsRequest, String.format("is all success：%s", batchGetRowResponse.isAllSucceed()));

            parseBatchRowRes(dpsRequest, tableName, primaryKeyName, resMap, batchGetRowResponse);
            if (!batchGetRowResponse.isAllSucceed()) {
                for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getFailedRows()) {
                    LogUtils.debug(dpsRequest, String.format("failure line：%s", batchGetRowRequest.getPrimaryKey(rowResult.getTableName(), rowResult.getIndex())));
                    LogUtils.debug(dpsRequest, String.format("failure reason：%s", rowResult.getError()));

                }
                //重试
                BatchGetRowRequest retryRequest = batchGetRowRequest.createRequestForRetry(batchGetRowResponse.getFailedRows());
                BatchGetRowResponse batchGetRowResponseAgain = client.batchGetRow(retryRequest);
                parseBatchRowRes(dpsRequest, tableName, primaryKeyName, resMap, batchGetRowResponseAgain);
            }
        } catch (Exception e) {
            LogUtils.error(dpsRequest, String.format("batchGetRowColumns,tableName:{}, primaryKeyName:{} occurs error", tableName, primaryKeyName, Throwables.getStackTraceAsString(e)));
        }
        return resMap;
    }

    /**
     * 批量获取指定列名
     *
     * @param dpsRequest
     * @param tableName
     * @param primaryKeyName
     * @param pkValues
     * @param columns
     * @param client
     * @return
     */
    public Map<String, Map<String, String>> batchGetRowColumns(DpsRequest dpsRequest, String tableName, String primaryKeyName, String primaryKeyName2,
                                                               List<Map<String, String>> pkValues, List<String> columns) {
        tableName = tableName + env;
        Map<String, Map<String, String>> resMap = new HashMap<>();
        if (CollectionUtils.isEmpty(pkValues)) return resMap;
        try {
            MultiRowQueryCriteria multiRowQueryCriteria = new MultiRowQueryCriteria(tableName);
            for (Map<String, String> itemId : pkValues) {
                PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
                primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName, PrimaryKeyValue.fromString(itemId.get(primaryKeyName)));
                primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName2, PrimaryKeyValue.fromString(itemId.get(primaryKeyName2)));
                PrimaryKey primaryKey = primaryKeyBuilder.build();
                multiRowQueryCriteria.addRow(primaryKey);
            }


            multiRowQueryCriteria.setMaxVersions(1);
            for (String column : columns) {
                multiRowQueryCriteria.addColumnsToGet(column);
            }
            BatchGetRowRequest batchGetRowRequest = new BatchGetRowRequest();
            batchGetRowRequest.addMultiRowQueryCriteria(multiRowQueryCriteria);
            BatchGetRowResponse batchGetRowResponse = client.batchGetRow(batchGetRowRequest);
            //LogUtils.debug(dpsRequest, String.format("is all success：%s", batchGetRowResponse.isAllSucceed()));

            parseBatchRowRes(dpsRequest, tableName, primaryKeyName, resMap, batchGetRowResponse);
            if (!batchGetRowResponse.isAllSucceed()) {
                for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getFailedRows()) {
                    LogUtils.debug(dpsRequest, String.format("failure line：%s", batchGetRowRequest.getPrimaryKey(rowResult.getTableName(), rowResult.getIndex())));
                    LogUtils.debug(dpsRequest, String.format("failure reason：%s", rowResult.getError()));

                }
                //重试
                BatchGetRowRequest retryRequest = batchGetRowRequest.createRequestForRetry(batchGetRowResponse.getFailedRows());
                BatchGetRowResponse batchGetRowResponseAgain = client.batchGetRow(retryRequest);
                parseBatchRowRes(dpsRequest, tableName, primaryKeyName, resMap, batchGetRowResponseAgain);
            }
        } catch (Exception e) {
            LogUtils.error(dpsRequest, String.format("batchGetRowColumns,tableName:{}, primaryKeyName:{} occurs error", tableName, primaryKeyName, Throwables.getStackTraceAsString(e)));
        }
        return resMap;
    }


    private static void parseBatchRowRes(DpsRequest dpsRequest, String tableName, String primaryKeyName, Map<String, Map<String, String>> resMap,
                                         BatchGetRowResponse batchGetRowResponse) {
        try {
            List<BatchGetRowResponse.RowResult> batchGetRowResult = batchGetRowResponse.getBatchGetRowResult(tableName);
            for (BatchGetRowResponse.RowResult rowResult : batchGetRowResult) {
                Map<String, String> tmpMap = new HashMap<>();
                Row row = rowResult.getRow();
                if (null == row) {
                    continue;
                }
                Column[] columns = row.getColumns();
                String primaryKey = row.getPrimaryKey().getPrimaryKeyColumn(primaryKeyName).getValue().toColumnValue().getValue().toString();
                for (Column column : columns) {
                    String name = column.getName();
                    String value = column.getValue().getValue().toString();
                    tmpMap.put(name, value);
                }
                resMap.put(primaryKey, tmpMap);
            }
        } catch (Exception e) {
            LogUtils.error(dpsRequest, String.format(Throwables.getStackTraceAsString(e)));
        }
    }

    private Map<String, String> getTsLayerTagMap(DpsRequest dpsRequest, String tableName) {
        Map<String, String> tagMapper = new HashMap<>();
        String sence = dpsRequest.getRequest().getSence();
        String layerName = dpsRequest.getLayer().getLayerName();
        tagMapper.put("sence", StringUtils.isEmpty(sence) ? "" : sence);
        tagMapper.put("layer", StringUtils.isEmpty(layerName) ? "" : layerName);
        tagMapper.put("tableName", StringUtils.isEmpty(tableName) ? "" : tableName);
        return tagMapper;
    }


    /**
     * 单行获取
     *
     * @param dpsRequest
     * @param pkValue        主键值
     * @param tableName      表名
     * @param primaryKeyName 主键名称
     * @param client
     * @return
     */
    @SneakyThrows
    public Map<String, String> getRow(DpsRequest dpsRequest, String pkValue, String tableName, String primaryKeyName) {
        CustomizedMetricsProcessor processor = CustomizedMetricsProcessor.builder("dps_sence_layer_ts").tagsMap(getTsLayerTagMap(dpsRequest, tableName)).build();
        return processor.process(() -> {
            Stopwatch timer = Stopwatch.createStarted();
            Map<String, String> resMap = new HashMap<>();
            try {
                if (StringUtils.isEmpty(pkValue)) return resMap;
                String tableNameStr = tableName + env;
                //构造主键。
                PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
                primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName, PrimaryKeyValue.fromString(pkValue));
                PrimaryKey primaryKey = primaryKeyBuilder.build();
                //读取一行数据，设置数据表名称
                SingleRowQueryCriteria criteria = new SingleRowQueryCriteria(tableNameStr, primaryKey);
                //设置读取最新版本。
                criteria.setMaxVersions(1);
                GetRowResponse getRowResponse = client.getRow(new GetRowRequest(criteria));
                Row row = getRowResponse.getRow();
                if (null == row) {
                    //LogUtils.warn(dpsRequest, String.format("tableName:%s,primaryKeyName:%s,pkValue:%s has not get info!", tableName, primaryKeyName, pkValue));
                    return resMap;
                }
                Column[] columns = row.getColumns();
                for (Column column : columns) {
                    resMap.put(column.getName(), column.getValue().getValue().toString());
                }
            } catch (Exception e) {
                LogUtils.error(dpsRequest, String.format("getRow,tableName:%s, primaryKeyName:%s occurs error:%s", tableName, primaryKeyName, Throwables.getStackTraceAsString(e)));
            } finally {
                timer.stop();
                LogUtils.info(dpsRequest, String.format("tableName=%s,primaryKeyName=%s,pkValue=%s,time=%s", tableName, primaryKeyName, pkValue, timer.elapsed(TimeUnit.MILLISECONDS)));
            }
            return resMap;
        });
    }


    /**
     * 单行获取指定列名
     *
     * @param dpsRequest
     * @param pkValue
     * @param tableName
     * @param primaryKeyName
     * @param cols
     * @param client
     * @return
     */
    @SneakyThrows
    public Map<String, String> getRowColumns(DpsRequest dpsRequest, String pkValue, String tableName, String primaryKeyName, List<String> cols) {
        CustomizedMetricsProcessor processor = CustomizedMetricsProcessor.builder("dps_sence_layer_ts").tagsMap(getTsLayerTagMap(dpsRequest, tableName)).build();
        return processor.process(() -> {
            Stopwatch timer = Stopwatch.createStarted();
            Map<String, String> resMap = new HashMap<>();
            if (StringUtils.isEmpty(pkValue)) return resMap;
            try {
                String tableNameStr = tableName + env;
                PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
                primaryKeyBuilder.addPrimaryKeyColumn(primaryKeyName, PrimaryKeyValue.fromString(pkValue));
                PrimaryKey primaryKey = primaryKeyBuilder.build();
                SingleRowQueryCriteria criteria = new SingleRowQueryCriteria(tableNameStr, primaryKey);
                criteria.setMaxVersions(1);
                for (String col : cols) {
                    criteria.addColumnsToGet(col);
                }
                GetRowResponse getRowResponse = client.getRow(new GetRowRequest(criteria));
                Row row = getRowResponse.getRow();
                if (null == row) {
                    LogUtils.warn(dpsRequest, String.format("the primaryKeyName:%s,pkValue:%s has not get info!", primaryKeyName, pkValue));
                    return resMap;
                }
                Column[] columns = row.getColumns();
                for (Column column : columns) {
                    resMap.put(column.getName(), column.getValue().getValue().toString());
                }
            } catch (Exception e) {
                LogUtils.error(dpsRequest, String.format("getRowColumns,tableName:{}, primaryKeyName:{} occurs error", tableName, primaryKeyName, e));
            } finally {
                timer.stop();
                LogUtils.info(dpsRequest, String.format("tableName=%s,primaryKeyName=%s,pkValue=%s,time=%s", tableName, primaryKeyName, pkValue, timer.elapsed(TimeUnit.MILLISECONDS)));
            }
            return resMap;
        });


    }

}
